package com.stu.rocketmq.syncmessage;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;

public class RocketMQConsumer {

    private static Logger logger = LoggerFactory.getLogger(RocketMQConsumer.class);

    public static void main(String[] args) throws MQClientException, IOException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sync-consumer-group");
        consumer.setNamesrvAddr("172.168.0.101:9876;172.168.0.103:9876");
        consumer.setInstanceName("sync-consumer-instance");
        consumer.subscribe("sync-message-topic","*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt msg : messageList) {
                    logger.info("消费者消费数据:{}", new String(msg.getBody()));
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        logger.info("{},开始消费消息",consumer.getInstanceName());
        consumer.start();
        System.in.read();
    }
}
